/************************************************************************/
// Copyright (C) 2016, han_gangbiao. All rights reserved.
// Use of this source code is governed by a GPL-v2 license that can be found in the License file.
// 
// [Time]:      2016-1-4 21:53:17
// [Author]:    han_gangbiao [http://my.oschina.net/jackhen]
// [Info]:      
/************************************************************************/

#include "net_reactor_impl.h"
#include "os_ns_sys_socket.h"
#include "net/net_reactor.h"

net_reactor_impl_t::net_reactor_impl_t()
	:m_pHandler(0)
{
	m_epfd = -1;
	m_ep_events = NULL;
    m_evt_idx = 0;
    m_evt_total = 0;
    m_thread_pool.clear();
}

net_reactor_impl_t::~net_reactor_impl_t()
{
	//netos::closesocket(m_epfd);
	//SAFE_ARRAY_DELETE(m_ep_events);
    uninit_reactor();
}

int net_reactor_impl_t::init_reactor( IEvent_handler* pHandler, int threadnum )
{
	if(threadnum <= 0) return -1;
    if(m_thread_pool.size() > 0)
        return 0;

	if(m_pHandler != NULL)
		return -1;

	m_pHandler = pHandler;

	m_epfd = epoll_create(EPOLL_MAX_FD_SIZE + EPOLL_RESERVED_FD_SIZE);
	if(m_epfd < 0)
		return -1;
	if(m_ep_events != NULL)
		SAFE_ARRAY_DELETE(m_ep_events);
	m_ep_events = new epoll_event_t[EPOLL_MAX_FD_SIZE];

	for(int i=0; i<threadnum; ++i)
	{
		m_thread_pool[i] = new smart::thread_t<net_reactor_impl_t>(&net_reactor_impl_t::running_task, this);
		if(m_thread_pool[i])
			m_thread_pool[i]->start();
	}

	return 0;
}

int net_reactor_impl_t::uninit_reactor()
{
    if(m_thread_pool.size() == 0)
        return 0;
    
	set_destroy_status(true);   //avoid to the threads busy-waiting

	{// free thread pool
		std::map<int, smart::threadbase_t*>::iterator it = m_thread_pool.begin();
		for(; it!=m_thread_pool.end(); ++it)
		{
			smart::threadbase_t* pThr = it->second;
			if(pThr) { pThr->stop(); SAFE_DELETE(pThr); }
		}
		m_thread_pool.clear();
	}
	m_pHandler = 0;

	
	del_all_listen_bean();
	del_all_conn_bean();

	netos::closesocket(m_epfd); m_epfd = -1;
	SAFE_ARRAY_DELETE(m_ep_events);
    m_evt_idx = m_evt_total = 0;

	net_buffer_pool_t::uninstance();
	return 0;
}

SOCK_FD net_reactor_impl_t::open_tcp_server( const char* local_ip, int local_port )
{// some services can be multiplex this reactor engine
	smart::ref_ptr<NRSocket> listen_bean = new NRSocket();
	int ret = listen_bean->open(AF_INET, SOCK_STREAM, IPPROTO_TCP, true);
	if(ret < 0)
		return -1;

	ret = listen_bean->set_nonblock();
	if(ret < 0)
		return -1;

	netaddr_t net_addr(local_ip, (uint16)local_port);
	if( netos::bind( listen_bean->get_sockfd(), (const struct sockaddr*)net_addr.get_addr(), net_addr.get_addr_len() ) < 0 )
	{
		listen_bean->close();
		return -1;
	}

	if( netos::listen( listen_bean->get_sockfd(), 5) < 0 )
	{
		listen_bean->close();
		return -1;
	}

	epoll_event_t ep_evt = {0};
	ep_evt.events = EPOLLIN; // (Level Triggered)
	ep_evt.data.fd = listen_bean->get_sockfd();//listen socket
	if( epoll_ctl(m_epfd, EPOLL_CTL_ADD, listen_bean->get_sockfd(), &ep_evt) < 0 )
	{
		listen_bean->close();
		return -1;
	}

	add_listen_bean(listen_bean.getPointer());

	return listen_bean->get_sockfd();
}

SOCK_FD net_reactor_impl_t::open_tcp_client( const char* remote_ip, int remote_port )
{// some clients can be multiplex this reactor engine
	smart::ref_ptr<NRSocket> cli_bean = new NRSocket();
	int ret = cli_bean->open(AF_INET, SOCK_STREAM, IPPROTO_TCP, true);
	if(ret < 0)
		return -1;

	netaddr_t net_addr(remote_ip, (uint16)remote_port);
	if( netos::connect( cli_bean->get_sockfd(), (const struct sockaddr*)net_addr.get_addr(), net_addr.get_addr_len() ) < 0 )
	{//block connect
		cli_bean->close();
		return -1;
	}

	cli_bean->set_nonblock();
	strncpy(cli_bean->m_ip, remote_ip, sizeof(cli_bean->m_ip));
	cli_bean->m_port = remote_port;

	epoll_event_t ep_evt = {0};
	ep_evt.events = EPOLLIN /*| EPOLLOUT*/ | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; //(Level Triggered),not EPOLLOUT
	ep_evt.data.fd = cli_bean->get_sockfd();//client socket
	if( epoll_ctl(m_epfd, EPOLL_CTL_ADD, cli_bean->get_sockfd(), &ep_evt) < 0 )
	{
		cli_bean->close();
		return -1;
	}

	add_conn_bean(cli_bean.getPointer());

	return cli_bean->get_sockfd();
}

SOCK_FD net_reactor_impl_t::open_udp_endpoint( const char* local_ip, int local_port )
{
	smart::ref_ptr<NRSocket> udp_bean = new NRSocket();
	int ret = udp_bean->open(AF_INET, SOCK_DGRAM, IPPROTO_UDP, true);
	if(ret < 0)
		return -1;

	netaddr_t net_addr(local_ip, (uint16)local_port);
	if( netos::bind( udp_bean->get_sockfd(), (const struct sockaddr*)net_addr.get_addr(), net_addr.get_addr_len() ) < 0 )
	{
		udp_bean->close();
		return -1;
	}

	udp_bean->set_nonblock();

	epoll_event_t ep_evt = {0};
	ep_evt.events = EPOLLIN/* | EPOLLOUT*/ | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; //(Level Triggered),not EPOLLOUT
	ep_evt.data.fd = udp_bean->get_sockfd();//udp socket
	if( epoll_ctl(m_epfd, EPOLL_CTL_ADD, udp_bean->get_sockfd(), &ep_evt) < 0 )
	{
		udp_bean->close();
		return -1;
	}

	add_conn_bean(udp_bean.getPointer());

	return udp_bean->get_sockfd();
}

int net_reactor_impl_t::send( SOCK_FD fd, const char* data, int len )
{
	smart::ref_ptr<NRSocket> conn_bean = get_conn_bean(fd);
	if(conn_bean == NULL)
	{
		printf("fd=%d,not found in ConnMap", fd);
		return -1;
	}

    //1> send to buffer
    conn_bean->lock_send(); // lock this bean's send buffer
	smart::ref_ptr<net_buffer_row_t> row = net_buffer_pool_t::instance()->create_buffer_row();
	row->m_fd = fd;
	row->m_data.assign(data, len);
	memset(row->m_ip, 0, sizeof(row->m_ip));
	strncpy(row->m_ip, conn_bean->m_ip, sizeof(row->m_ip));
	row->m_port = conn_bean->m_port;

	conn_bean->push_row_to_send_buffer_back(row.getPointer());

    //2> send buffer's data
    {
        int send_len = 0;
        int send_result = 0;
        int stop = conn_bean->send_buffer_size();
        while(stop > 0)
        {
            smart::ref_ptr<net_buffer_row_t> buf_row = conn_bean->pop_row_from_send_buffer_front();
            if( conn_bean->get_sock_type() == SOCK_STREAM )
                send_result = do_send(conn_bean->get_sockfd(), buf_row->m_data.c_str(), buf_row->m_data.length(), send_len);
            else if( conn_bean->get_sock_type() == SOCK_DGRAM )
                send_result = do_sendto(conn_bean->get_sockfd(), buf_row->m_data.c_str(), buf_row->m_data.length(), buf_row->m_ip, buf_row->m_port, send_len);
            if(send_result < 0) //Error or EAGAIN
            {
                if(send_len > 0 && send_len < buf_row->m_data.length())
                {
                    const char* tmp_data = buf_row->m_data.c_str();
                    int tmp_len = buf_row->m_data.length();
                    buf_row->m_data.assign(tmp_data + send_len, tmp_len - send_len);
                    conn_bean->push_row_to_send_buffer_front(buf_row.getPointer());
                    break;
                }
                else if(send_len == 0)
                {
                    conn_bean->push_row_to_send_buffer_front(buf_row.getPointer());
                    break;
                }
            }
            else
            {//recyle to buffer_pool
                net_buffer_pool_t::instance()->recycle(buf_row);
            }
            stop--;
        }

        //3> is send over?
        if(stop > 0)
        {
            epoll_event_t ep_evt = {0};
            ep_evt.events = EPOLLIN | EPOLLOUT | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; // (Level Triggered)
            ep_evt.data.fd = conn_bean->get_sockfd();
            epoll_ctl(m_epfd, EPOLL_CTL_MOD, conn_bean->get_sockfd(), &ep_evt);
        }
    }
    conn_bean->unlock_send(); // unlock
    
	return 0;
}

int net_reactor_impl_t::sendto( SOCK_FD fd, const char* data, int len, const char* remote_ip, int remote_port )
{
	smart::ref_ptr<NRSocket> conn_bean = get_conn_bean(fd);
	if(conn_bean == NULL)
	{
		printf("fd=%d,not found in ConnMap", fd);
		return -1;
	}

	smart::ref_ptr<net_buffer_row_t> row = net_buffer_pool_t::instance()->create_buffer_row();
	row->m_fd = fd;
	row->m_data.assign(data, len);
	memset(row->m_ip, 0, sizeof(row->m_ip));
	strncpy(row->m_ip, remote_ip, sizeof(row->m_ip));
	row->m_port = remote_port;

	conn_bean->push_row_to_send_buffer_back(row.getPointer());
	return 0;
}

int net_reactor_impl_t::close_fd( SOCK_FD fd )
{
	smart::guard_t guard(&m_mtx_pending_close);
	// maybe listen sock or conn sock
	m_pending_close_fd_map[fd] = NET_ERR_USER_CLOSED; // user request close this socket
	return 0;
}

int net_reactor_impl_t::running_task( smart::threadbase_t* thr )
{
    m_thread_pool_mtx.lock();
    
    if(deal_pending_destroy() < 0)
    {//user calling uninit_reactor
        m_thread_pool_mtx.unlock(); //must unlock
        return -1; //exit this thread
    }
    
    deal_pending_close_fdQ(); //deal with pending close sockets
    
    if( m_evt_idx == m_evt_total ) {
        m_evt_idx = m_evt_total = 0;
        int timeout = 100; //100ms
        int ret = epoll_wait(m_epfd, m_ep_events, EPOLL_MAX_FD_SIZE, timeout);
        if( ret <= 0 ) {
            m_thread_pool_mtx.unlock();
            return 0;
        }
        m_evt_total = ret;
    }
    
    // poll a event
    smart::ref_ptr<NRSocket> bean = NULL;
    uint32 evt = 0;
    for(; m_evt_idx < m_evt_total; ++m_evt_idx) {
        SOCK_FD fd = m_ep_events[m_evt_idx].data.fd;
        if( is_listen_bean(fd) == true ) {
            bean = get_listen_bean(fd);
        } else {
            bean = get_conn_bean(fd);
        }
        if(bean == NULL) continue;
        evt = m_ep_events[m_evt_idx].events;
        m_evt_idx++;
        break;
    }
    m_thread_pool_mtx.unlock();
    
    //>deal event
    if(bean == NULL)
    {//maybe bean is null
        printf("bean == NULL!\n");
        return 0;
    }
    if( is_listen_bean(bean->get_sockfd()) == true ) {
        while( do_accept(bean->get_sockfd()) > 0 );
    } else {
        if( (evt & EPOLLIN) != 0 && (evt & EPOLLRDHUP) ) {
            printf("EPOLLIN && EPOLLRDHUP tid[%u]\n", thr->get_tid());
			do_close(bean->get_sockfd(), NET_ERR_PEER_CLOSED);
            return 0;
        }
        if( (evt & EPOLLERR) != 0 && (evt & EPOLLHUP) ) {
            printf("EPOLLERR && EPOLLHUP\n");
			do_close(bean->get_sockfd(), NET_ERR_LOCAL_WRONG);
            return 0;
        }
        if( (evt & EPOLLIN) != 0 || (evt & EPOLLPRI) != 0 ) { //read
            bean->lock_recv(); //lock bean
            //> recv to buffer
            if( bean->get_sock_type() == SOCK_STREAM )
				do_recv(bean->get_sockfd(), bean->m_ip, bean->m_port);
			else if( bean->get_sock_type() == SOCK_DGRAM )
				do_recvfrom(bean->get_sockfd());
            //> callback to user
            int stop = bean->recv_buffer_size();
            for(int i=0; i<stop; ++i)
            {
                smart::ref_ptr<net_buffer_row_t> row = bean->pop_row_from_recv_buffer_front();
                if(m_pHandler)
                    m_pHandler->on_data(bean->get_sockfd(), row->m_data.c_str(), row->m_data.length(), row->m_ip, row->m_port);
                net_buffer_pool_t::instance()->recycle(row);
            }
            bean->unlock_recv(); //unlock bean
        }
        if( (evt & EPOLLOUT) != 0 ) { //write
            bean->lock_send(); //lock bean
            int send_len = 0;
            int send_result = 0;
			int stop = bean->send_buffer_size();
			while(stop > 0)
			{
				smart::ref_ptr<net_buffer_row_t> buf_row = bean->pop_row_from_send_buffer_front();
				if( bean->get_sock_type() == SOCK_STREAM )
					send_result = do_send(bean->get_sockfd(), buf_row->m_data.c_str(), buf_row->m_data.length(), send_len);
				else if( bean->get_sock_type() == SOCK_DGRAM )
					send_result = do_sendto(bean->get_sockfd(), buf_row->m_data.c_str(), buf_row->m_data.length(), buf_row->m_ip, buf_row->m_port, send_len);
				if(send_result < 0) // Error or EAGAIN
				{
					if(send_len > 0 && send_len < buf_row->m_data.length())
					{
						const char* tmp_data = buf_row->m_data.c_str();
						int tmp_len = buf_row->m_data.length();
						buf_row->m_data.assign(tmp_data + send_len, tmp_len - send_len);
						bean->push_row_to_send_buffer_front(buf_row.getPointer());
						break;
					}
                    else if(send_len == 0)
                    {
                        bean->push_row_to_send_buffer_front(buf_row.getPointer());
                        break;
                    }
				}
				else
                {//recyle to buffer_pool
                    net_buffer_pool_t::instance()->recycle(buf_row);
				}
				stop--;
			}//~while(stop > 0)
            
            //>is send over?
            if(stop <= 0)
            {//移除EPOLLOUT事件 //TODO:fix 需要判断是否已经发送完了,否则有剩余数据则无法触发继续发送了; 如果还有缓存,则继续保留EPOLLOUT
                epoll_event_t ep_evt = {0};
                ep_evt.events = EPOLLIN/* | EPOLLOUT*/ | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; //(Level Triggered)
                ep_evt.data.fd = bean->get_sockfd();
                epoll_ctl(m_epfd, EPOLL_CTL_MOD, bean->get_sockfd(), &ep_evt);
            }
            bean->unlock_send(); //unlock bean
        }
    }
    return 0;
}

int net_reactor_impl_t::deal_pending_close_fdQ()
{
    //smart::guard_t guard(&m_mtx_pending_close);
	m_mtx_pending_close.lock();
    std::map<SOCK_FD, int>::iterator it = m_pending_close_fd_map.begin();
    for(; it != m_pending_close_fd_map.end(); ++it)
    {
        SOCK_FD fd = it->first;

		epoll_event_t ep_evt = {0};
		//TODO: EPOLL_CTL_DEL!!!
		//注: 在2.6.9以前的内核中,当执行EPOLL_CTL_DEL时,event须非空,但在之后,event可空,一般为了兼容以前的内核版本,最好将event非空.
		ep_evt.events = EPOLLIN | EPOLLOUT | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; //stream_sock

        smart::ref_ptr<NRSocket> bean = get_conn_bean(fd);
        if(bean == NULL)
        {// listen sock
        	bean = get_listen_bean(fd);
			ep_evt.events = EPOLLIN;
        }
		else
		{
			if(bean->get_sock_type() == SOCK_DGRAM)
			{
				ep_evt.events = EPOLLIN | EPOLLOUT | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; //(Level Triggered)
			}
		}
        if(bean == NULL)
        {//should not be found
        	continue;
        }

		ep_evt.data.fd = bean->get_sockfd();
		epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, &ep_evt);

        if(m_pHandler)
		{
			m_mtx_pending_close.unlock();
			m_pHandler->on_close(fd, it->second);
			m_mtx_pending_close.lock();
		}

		del_listen_bean(fd); //tcp-server will not receive a new connect.
		del_conn_bean(fd);
    }
    m_pending_close_fd_map.clear();

	m_mtx_pending_close.unlock();
    return 0;
}

void net_reactor_impl_t::set_destroy_status( bool status )
{
	smart::guard_t guard(&m_mtx_pending_destroy);
	m_pending_destroy = status;
}

int net_reactor_impl_t::deal_pending_destroy()
{
	smart::guard_t guard(&m_mtx_pending_destroy);
	if(m_pending_destroy)
		return -1;
	return 0;
}

int net_reactor_impl_t::do_accept( SOCK_FD listen_fd )
{
    int ret_fd = -1;
    struct sockaddr_in addr;
    int addr_len = sizeof(struct sockaddr_in);
    SOCK_FD cli_fd = netos::accept(listen_fd, (struct sockaddr*)&addr, &addr_len);
    ret_fd = cli_fd;
    if(cli_fd < 0) {
        if(cli_fd == -1) {
            if(errno != EAGAIN && errno != ECONNABORTED && errno != EPROTO && errno != EINTR)
                printf("accept error!\n");
        }
        return cli_fd;
    }
    smart::ref_ptr<NRSocket> bean = new NRSocket();
    bean->set_sockfd( cli_fd ); bean->set_sock_type( SOCK_STREAM );
    if(bean->is_valid())
    {
        const char* remote_ip = inet_ntoa(addr.sin_addr);
        int remote_port = ntohs(addr.sin_port);
        memset(bean->m_ip, 0, sizeof(bean->m_ip));
        strncpy(bean->m_ip, remote_ip, sizeof(bean->m_ip));
        bean->m_port = remote_port;

        int subRet = 0;
        if(m_pHandler)
            subRet = m_pHandler->on_connect(listen_fd, bean->get_sockfd(), remote_ip, remote_port);

        if(subRet >= 0)
        {
            add_conn_bean(bean.getPointer());
            //set socket buffer 48KB (default is 8KB)
            bean->set_sock_recv_buf(SOCK_BUFFER_SIZE/4*3);
            bean->set_sock_send_buf(SOCK_BUFFER_SIZE/4*3);
            bean->set_nonblock();
            bean->set_addr_reuse();
            
            epoll_event_t ep_evt = {0};
            ep_evt.events = EPOLLIN /*| EPOLLOUT*/ | EPOLLPRI | EPOLLRDHUP | EPOLLERR | EPOLLHUP; //(Edge Triggered)
            ep_evt.data.fd = bean->get_sockfd();
            epoll_ctl(m_epfd, EPOLL_CTL_ADD, bean->get_sockfd(), &ep_evt);
        }
        else
        {//user reject this connect.
            bean->close();
        }
    }
    return ret_fd;
}

int net_reactor_impl_t::do_recv( SOCK_FD fd, const char*ip, int port )
{
    int result = 0;
    int recv_len = 0;
    char *recv_buffer = new char[SOCK_BUFFER_SIZE/2]; //dynamic memory
    if((recv_len = netos::recv(fd, recv_buffer, SOCK_BUFFER_SIZE/2)) > 0) {
        smart::ref_ptr<NRSocket> bean = get_conn_bean(fd);
        if(bean == NULL) {
            return -1;
        }
        smart::ref_ptr<net_buffer_row_t> row = net_buffer_pool_t::instance()->create_buffer_row();
        row->m_data.assign(recv_buffer, recv_len);

        memset(row->m_ip, 0, sizeof(row->m_ip));
        strncpy(row->m_ip, ip, sizeof(row->m_ip));
        row->m_port = port;
        row->m_fd = fd;

        bean->push_row_to_recv_buffer_back(row.getPointer());

        result = recv_len;
    }
    if(recv_len == -1 && errno != EAGAIN)
    {//net error
        do_close(fd, NET_ERR_RECV_CLOSING);
        result = -1;
    }
    else if(recv_len == 0)
    {//conn closed
        do_close(fd, NET_ERR_RECV_CLOSING);
        result = -1;
    }
    else if(recv_len == -1 && errno == EAGAIN)
    {
        result = 0;
    }
    SAFE_ARRAY_DELETE(recv_buffer); //free memory
    return result;
}

int net_reactor_impl_t::do_recvfrom( SOCK_FD fd )
{
    int result = 0;
	int recv_len = 0;
    char *recv_buffer = new char[SOCK_BUFFER_SIZE/2]; //dynamic memory

	struct sockaddr_in from;
	int from_len = sizeof(struct sockaddr_in);
	recv_len = netos::recvfrom(fd, recv_buffer, SOCK_BUFFER_SIZE/2, 0, (struct sockaddr*)&from, &from_len);
	if(recv_len <= 0 && errno != EAGAIN)
	{//error
		do_close(fd, NET_ERR_RECV_CLOSING);
        result = -1;
	}
    else if(recv_len == -1 && errno == EAGAIN)
    {
        result = 0;
    }
	else
	{
		smart::ref_ptr<NRSocket> bean = get_conn_bean(fd);
		if(bean == NULL)
			return -1;

		smart::ref_ptr<net_buffer_row_t> row = net_buffer_pool_t::instance()->create_buffer_row();
		row->m_data.assign(recv_buffer, recv_len);

		row->m_port = ntohs(from.sin_port);
		const char* ip = inet_ntoa(from.sin_addr);
		memset(row->m_ip, 0, sizeof(row->m_ip));
		strncpy(row->m_ip, ip, sizeof(row->m_ip));
		row->m_fd = fd;

		bean->push_row_to_recv_buffer_back(row.getPointer());

        result = recv_len;
	}

    SAFE_ARRAY_DELETE(recv_buffer); //free memory
	return result;
}

int net_reactor_impl_t::do_send( SOCK_FD fd, const char* data, int len, int& send_len )
{
    int result = 0;
    int tmp_len = 0;
    int ntotal = len;
    while(ntotal > 0) {
        tmp_len = netos::send(fd, data+len-ntotal, ntotal);
        if( tmp_len == -1 && errno == EAGAIN ) {
            result = -1;
            break;
        } else if( (tmp_len <=0 && tmp_len != -1) || (tmp_len == -1 && errno != EAGAIN) ) { //should do_close
            do_close(fd, NET_ERR_SEND_CLOSING);
            result = -1;
            break;
        }
        ntotal -= tmp_len;
    }

    send_len = len - ntotal;
    {
        if(m_pHandler)
            m_pHandler->on_send_ack(fd, send_len);
    }
    return result;
}

int net_reactor_impl_t::do_sendto( SOCK_FD fd, const char* data, int len, const char* remote_ip, int remote_port, int& send_len )
{
    struct sockaddr_in to;
    int to_len = sizeof(struct sockaddr_in);

    memset(&to, 0, sizeof(to));
    to.sin_family = AF_INET;
    to.sin_addr.s_addr = inet_addr(remote_ip);
    to.sin_port = htons(remote_port);

    int result = 0;
    int tmp_len = 0;
    int ntotal = len;
    while(ntotal > 0) {
        tmp_len = netos::sendto(fd, data+len-ntotal, ntotal, 0, (const struct sockaddr*)&to, to_len);
        if( tmp_len == -1 && errno == EAGAIN ) {
            result = -1;
            break;
        } else if( (tmp_len <=0 && tmp_len != -1) || (tmp_len == -1 && errno != EAGAIN) ) { //should do_close
            do_close(fd, NET_ERR_SEND_CLOSING);
            result = -1;
            break;
        }
        ntotal -= tmp_len;
    }

    send_len = len - ntotal;
    {
        if(m_pHandler)
            m_pHandler->on_send_ack(fd, send_len);
    }
    return result;
}

int net_reactor_impl_t::do_close( SOCK_FD fd, int err )
{
    smart::guard_t guard(&m_mtx_pending_close);
    m_pending_close_fd_map[fd] = err;

    return 0;
}

int net_reactor_impl_t::add_conn_bean( NRSocket* bean )
{
	smart::guard_t guard(&m_mtx_conn_map);
	NetConnMap::iterator it = m_conn_map.find(bean->get_sockfd());
	if(it != m_conn_map.end())
	{
		NRSocket* tmp = (it->second).getPointer();
		tmp->close();
		it->second = NULL;
		m_conn_map.erase(it);
	}
	m_conn_map[bean->get_sockfd()] = bean;
	return 0;
}

int net_reactor_impl_t::del_conn_bean( SOCK_FD fd )
{
	smart::guard_t guard(&m_mtx_conn_map);
	NetConnMap::iterator it = m_conn_map.find(fd);
	if(it != m_conn_map.end())
	{
		NRSocket* bean = (it->second).getPointer();
		bean->close();
		it->second = NULL;
		m_conn_map.erase(it);
		return 0;
	}
	return -1;
}

int net_reactor_impl_t::del_all_conn_bean()
{
	smart::guard_t guard(&m_mtx_conn_map);
	NetConnMap::iterator it = m_conn_map.begin();
	for(; it!=m_conn_map.end(); ++it)
	{
		NRSocket* bean = (it->second).getPointer();
		bean->close();
		it->second = NULL;
	}
	m_conn_map.clear();
	return 0;
}

smart::ref_ptr<NRSocket> net_reactor_impl_t::get_conn_bean( SOCK_FD fd )
{
	NRSocket* bean = NULL;
	smart::guard_t guard(&m_mtx_conn_map);
	NetConnMap::iterator it = m_conn_map.find(fd);
	if(it != m_conn_map.end())
	{
		bean = (it->second).getPointer();
	}
	return bean;
}

int net_reactor_impl_t::add_listen_bean( NRSocket* bean )
{
	smart::guard_t guard(&m_mtx_listen_map);
	NetListenMap::iterator it = m_listen_map.find(bean->get_sockfd());
	if(it != m_listen_map.end())
	{
		NRSocket* tmp = (it->second).getPointer();
		tmp->close();
		it->second = NULL;
		m_listen_map.erase(it);
	}
	m_listen_map[bean->get_sockfd()] = bean;
	return 0;
}

int net_reactor_impl_t::del_listen_bean( SOCK_FD fd )
{
	smart::guard_t guard(&m_mtx_listen_map);
	NetListenMap::iterator it = m_listen_map.find(fd);
	if(it != m_listen_map.end())
	{
		NRSocket* bean = (it->second).getPointer();
		bean->close();
		it->second = NULL;
		m_listen_map.erase(it);
		return 0;
	}
	return -1;
}

int net_reactor_impl_t::del_all_listen_bean()
{
	smart::guard_t guard(&m_mtx_listen_map);
	NetListenMap::iterator it = m_listen_map.begin();
	for(; it!=m_listen_map.end(); ++it)
	{
		NRSocket* bean = (it->second).getPointer();
		bean->close();
		it->second = NULL;
	}
	m_listen_map.clear();
	return 0;
}

smart::ref_ptr<NRSocket> net_reactor_impl_t::get_listen_bean( SOCK_FD fd )
{
	NRSocket* bean = NULL;
	smart::guard_t guard(&m_mtx_listen_map);
	NetListenMap::iterator it = m_listen_map.find(fd);
	if(it != m_listen_map.end())
	{
		bean = (it->second).getPointer();
	}
	return bean;
}

bool net_reactor_impl_t::is_listen_bean( SOCK_FD fd )
{
	smart::guard_t guard(&m_mtx_listen_map);
	NetListenMap::iterator it = m_listen_map.find(fd);
	if(it != m_listen_map.end())
		return true;
	return false;
}



